package yuekao5.util;

import com.alibaba.fastjson.JSON;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import yuekao5.entity.Tm6;

import java.sql.*;
import java.sql.DriverManager;
import java.util.Collections;

public class AsyncIOUtil extends RichAsyncFunction<Tm6,Tm6> {
    Connection conn = null;
    @Override
    public void open(Configuration parameters) throws Exception {
        Class.forName("org.apache.phoenix.jdbc.PhoenixDriver");
        conn = DriverManager.getConnection("jdbc:phoenix:hadoop-single:2181", "", "");
        conn.setAutoCommit(true);
    }

    @Override
    public void close() throws Exception {
        conn.close();
    }


    @Override
    public void asyncInvoke(Tm6 tm6, ResultFuture<Tm6> resultFuture) throws Exception {
        PreparedStatement ps = conn.prepareStatement("select * from dim_base_region_info where id=?");
        ps.setObject(1,tm6.getSender_city_id());
        ResultSet rs = ps.executeQuery();
        while(rs.next()){
            String line = rs.getString("line");
            String name = JSON.parseObject(line).getJSONObject("data").getString("name");
            tm6.setName(name);
        }
        resultFuture.complete(Collections.singleton(tm6));
    }
}
